package com.surfilter.massdata.spark.util;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.surfilter.massdata.spark.bean.IpBean;
import com.surfilter.massdata.spark.util.ConvertUtil;
import com.surfilter.massdata.spark.util.IpCheckConstants;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class iPCheckZjUtil {
	
	private static final Log log = LogFactory.getLog(iPCheckZjUtil.class);
	
	
	/**
	 * 
	 * @Description:将rdd进行过滤，当flag是true返回正确的IpBean的rdd,为flase返回错误的IpBean的rdd
	 * @param ipBeanRdd
	 * @param flag
	 * @return
	 */
	public static JavaRDD<IpBean> getFilterIpBeanRdd(JavaRDD<IpBean> ipBeanRdd) {
		JavaRDD<IpBean> rdd = ipBeanRdd.filter(new Function<IpBean, Boolean>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Boolean call(IpBean ipBean) throws Exception {
				boolean flag=false;
				String tf=ipBean.getTflag();
				if(tf.equals(IpCheckConstants.IP_ALLOCATE_TYPE)){
					boolean resultFlag = validateIpProperties(ipBean);
					flag=resultFlag;
				}
				return flag;
				
			}
		});
		return rdd;
	}
	
	/**
	 * @Description:校验IP属性
	 * @param ipBean
	 * @return
	 */
	public static boolean validateIpProperties(IpBean ipBean) {
		boolean flag = true;
		try {
			long qsip = ipBean.getQsip();
			long zzip = ipBean.getZzip();
			IpUtils.longToIP(zzip);
			IpUtils.longToIP(qsip);
			// 然后监测IP段是否合法
			if (qsip > zzip) {
				return false;
			}
			// 接着监测IP段是否过大
			long ipSegMaxLen = Long.parseLong(Global.getConfig("surfilter.ipcheck.seg.len.max", "20000000"));
			if (zzip - qsip >= ipSegMaxLen) {
				return false;
			}
			// 如果前面都满足，继续校验，监测IP分配属性是否正确
			if (ipBean.getTflag().equals(IpCheckConstants.IP_ALLOCATE_TYPE)) {
				if (ipBean.getFpdx() == -1L || ipBean.getFpdx().equals(new Long(0l))) {
					if (ipBean.getFpfs().equals(new Long(2l))) {
						return false;
					}
				} else {
					if (!ipBean.getFpfs().equals(new Long(2l))) {
						return false;
					}
				}
			}
		} catch (Exception e) {
			flag = false;
		}
		return flag;
	}
	
	
	/**
	 * @Description 获取BR3010表中的id
	 * @param shengId
	 * @return
	 */
	public static  HashSet<Long>  getBr3010IdMap(String shengId) {
		HashSet<Long> br3010Set = Sets.newHashSet();
		String query_br3010_sql = "select id from br3010 where shengid='"+shengId+"'";
		Connection conn = JdbcUtilNew.getConnection();
		PreparedStatement ps = null;
		try {
			ps = conn.prepareStatement(query_br3010_sql);
			ResultSet rs = ps.executeQuery();
			while (rs.next()) {
				br3010Set.add(rs.getLong(1));
			}
			if (rs != null) {
				rs.close();
			}
		} catch (SQLException e) {
			e.printStackTrace();
		} finally {
			try {
				if (ps != null)
					ps.close();
				if (conn != null)
					conn.close();
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
		return br3010Set;
	}
	
	
	//获取BR3010表中的ID,DWMC,JYXKZBH
	public static Map<Long,String> getBR3010Map(){
		Map<Long, String> br3010Map = Maps.newHashMap();
		Connection conn = JdbcUtil.getConnection();
		PreparedStatement ps = null;
		String querySql = "select ID,DWMC,JYXKZBH from BR3010";
		try {
			ps = conn.prepareStatement(querySql);
			ResultSet rs = ps.executeQuery();
			while(rs.next()){
				Long id = rs.getLong(1);
				String bwmc = rs.getString(2);
				String jyxkzbh = rs.getString(3);
				br3010Map.put(id,bwmc+"|"+jyxkzbh);
			}
			if (rs != null){
				rs.close();
			}
		} catch (SQLException e) {
			if (log.isErrorEnabled()) {
				log.error(e.getMessage(), e);
			}
		} finally {
			try {
				if (ps != null)
					ps.close();
				if (conn != null)
					conn.close();
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
		return br3010Map;
		
	}
	
	//获取SY0002表中的CODE_KEY,CODE_VALUE
	public static Map<String, String> getSY0002Map(){
		Map<String, String> SY0002Map = Maps.newHashMap();
		Connection conn = JdbcUtil.getConnection();
		PreparedStatement ps = null;
		String querySql = "select CODE_KEY,CODE_VALUE from SY0002 where CODE_TYPE='OPERATORS_TYPE' and IS_OPERATE=1";
		try {
			ps = conn.prepareStatement(querySql);
			ResultSet rs = ps.executeQuery();
			while(rs.next()){
				String code_key = rs.getString(1);
				String code_value = rs.getString(2);
				SY0002Map.put(code_key,code_value);
			}
			if (rs != null){
				rs.close();
			}
		} catch (SQLException e) {
			if (log.isErrorEnabled()) {
				log.error(e.getMessage(), e);
			}
		} finally {
			try {
				if (ps != null)
					ps.close();
				if (conn != null)
					conn.close();
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
		return SY0002Map;
	}
	
	
	
	/**
	 * ip合并
	 * @param list
	 * @return
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public static List<Map<String,Object>> getIpSeg(Iterable<String> tuple2){
		List<Map<String,Object>> myList=new ArrayList<Map<String,Object>>();
		List<Map<String,Object>> result = new ArrayList<Map<String,Object>>();
		
		for (String str : tuple2) {
			Map<String,Object> ObjectList = Maps.newHashMap();
			String[] array = str.split("\001");
			ObjectList.put("time",Long.parseLong(array[0]));
			ObjectList.put("ip",Long.parseLong(array[1]));
			ObjectList.put("bean",array[2]);
			myList.add(ObjectList);
		}
		Collections.sort(myList, new  Comparator<Map>(){
			
			@Override
			public int compare(Map o1, Map o2) {
				Long ip1 = (Long) o1.get("ip");
				Long ip2 = (Long) o2.get("ip");
				return ip1.compareTo(ip2);
			}
		});
		//List<String> segStr = new ArrayList<String>();
		Long maxTime = 0L;
		String maxIpBean = null;
		int i = 0;
		long qsip = 0, zzip = 0;
		for(Map<String,Object> map : myList){
			if(null!=map&&map.size()>0){
				IpBean copyBean = new IpBean();
				if(myList.size()==1){
					//segStr.add(map.get("ip")+","+map.get("ip"));
					//给剩下的bean赋值
					ConvertUtil.convertIpBean(String.valueOf(map.get("bean")).split("\t"), copyBean);
					Map<String,Object> mapresult = Maps.newHashMap();
					mapresult.put("qsip",map.get("ip"));
					mapresult.put("zzip",map.get("ip"));
					mapresult.put("ipBean",copyBean);
					result.add(mapresult);
				}else{
					//Arrays.sort(array);
					long ip = (long) map.get("ip");
					Long currTime = (Long) map.get("time");
					String currBean = (String)map.get("bean");
					if (i == 0) {
						qsip = ip;
						zzip = ip;
						//设置初始值
						maxTime = currTime;
						maxIpBean = currBean;
						
						//给剩下的bean赋值
					}else if(i==myList.size()-1){
						if(ip==zzip+1){
							//连续的
							zzip=ip;
							//segStr.add(qsip + "," + zzip);
							
							if(maxTime < currTime){
								maxTime = currTime;
								maxIpBean = currBean;
							}
							ConvertUtil.convertIpBean(maxIpBean.split("\t"), copyBean);
							Map<String,Object> mapresult = Maps.newHashMap();
							mapresult.put("qsip",qsip);
							mapresult.put("zzip",zzip);
							mapresult.put("ipBean",copyBean);
							result.add(mapresult);
							
						}else{
							//segStr.add(qsip + "," + zzip);
							ConvertUtil.convertIpBean(maxIpBean.split("\t"), copyBean);
							Map<String,Object> mapresult = Maps.newHashMap();
							mapresult.put("qsip",qsip);
							mapresult.put("zzip",zzip);
							mapresult.put("ipBean",copyBean);
							result.add(mapresult);
							
							
							//segStr.add(ip + "," + ip);
							ConvertUtil.convertIpBean(currBean.split("\t"), copyBean);
							Map<String,Object> mapresult1 = Maps.newHashMap();
							mapresult1.put("qsip",ip);
							mapresult1.put("zzip",ip);
							mapresult1.put("ipBean",copyBean);
							result.add(mapresult1);
						}
					} else {
						{
							if (ip == zzip + 1) {
								zzip = ip;
								if(maxTime < currTime){
									maxTime = currTime;
									maxIpBean = currBean;
								}
							} else {
								//segStr.add(qsip + "," + zzip);
								ConvertUtil.convertIpBean(maxIpBean.split("\t"), copyBean);
								Map<String,Object> mapresult = Maps.newHashMap();
								mapresult.put("qsip",qsip);
								mapresult.put("zzip",zzip);
								mapresult.put("ipBean",copyBean);
								result.add(mapresult);
								
								qsip = ip;
								zzip = ip;
								//设置初始值
								maxTime = currTime;
								maxIpBean = currBean;
							}
						}
					}
				}
				++i;
			}
		}
	 return result;
 }
	
	
	
	/**
	 * @Description:将DataFrame转换成javaRdd<IpBean>
	 * @param resultFrame
	 * @return
	 */
	public static JavaRDD<IpBean> getIpBeanRdd(DataFrame resultFrame) {
		JavaRDD<IpBean> rdd = resultFrame.toJavaRDD().map(new Function<Row, IpBean>() {
			private static final long serialVersionUID = 1L;
			@Override
			public IpBean call(Row row) throws Exception {
				IpBean bean = new IpBean();
				bean.setStorageType(row.getString(0));
				if (bean.getStorageType().equals(IpCheckConstants.OLD_BASE)) {//5
					// 基准库
					bean.setId(( (BigDecimal) row.get(1)).longValue());
					bean.setQsip(null == row.get(2) ? null :((BigDecimal) row.get(2)).longValue());
					bean.setZzip(null == row.get(3) ? null : ((BigDecimal) row.get(3)).longValue());
					bean.setBbdw_id(null == row.get(4) ? null : Long.valueOf(row.get(4).toString()));
					bean.setSyfs(null == row.get(5) ? -1L : Long.valueOf(row.get(5).toString()));
					bean.setShengid(null == row.get(6) ? -1L : Long.valueOf(row.get(6).toString()));
					bean.setShiid(null == row.get(7) ? -1L : Long.valueOf(row.get(7).toString()));
					bean.setXianid(null == row.get(8) ? -1L : Long.valueOf(row.get(8).toString()));
					bean.setHouseCode(null == row.get(9) ? "N/A" :row.getString(9));
					bean.setHouseName(null == row.get(10) ? "N/A" : row.getString(10));
					bean.setFpfs(null == row.get(11) ? -1L : Long.valueOf(row.get(11).toString()));
					bean.setDwmc(null == row.get(12) ? null : row.getString(12));
					bean.setLxrxm(null == row.getString(13) ? "N/A" : row.getString(13));
					bean.setLxrdh(null == row.getString(14) ? "N/A" : row.getString(14));
					bean.setDwxz(null == row.get(15) ? null : Long.valueOf(row.get(15).toString()));
					bean.setDwhyfl(null == row.get(16) ? -1L : Long.valueOf(row.get(16).toString()));
					bean.setBz(null == row.get(17) ? null : row.getString(16));
					bean.setSourceIdcId(null == row.getString(18) ? null : row.getString(18));
					bean.setOperatorsId(null == row.getString(19) ? "N/A" : row.getString(19));
					bean.setOperatorsName(null == row.getString(20) ? "N/A" : row.getString(20));
					bean.setOperatorsBh(null == row.getString(21) ? "N/A" : row.getString(21));
					bean.setXxdz(null == row.getString(22) ? "N/A" : row.getString(22));
					bean.setJtiplx(null == row.get(23) ? -1 : Integer.valueOf(row.get(23).toString()));
					
				} else if (bean.getStorageType().equals(IpCheckConstants.BEIAN)) {//1
					// 备案库
					bean.setTflag(null == row.getString(1) ? null : row.getString(1));
					bean.setId(( (BigDecimal) row.get(2)).longValue());
					bean.setBbdw_id(null == row.get(3) ? null : Long.valueOf(row.get(3).toString()));
					bean.setSbsj(null == row.get(4) ? null : row.getTimestamp(4).getTime());
					bean.setQsip(null == row.get(5) ? null :Long.valueOf(row.get(5).toString()));
					bean.setZzip(null == row.get(6) ? null : Long.valueOf(row.get(6).toString()));
					bean.setSfgn(null == row.get(7) ? -1L : Long.valueOf(row.get(7).toString()));
					bean.setFpfs(null == row.get(8) ? -1L : Long.valueOf(row.get(8).toString()));
					bean.setFpdx(null == row.get(9) ? -1L : Long.valueOf(row.get(9).toString()));
					bean.setSyfs(null == row.get(10) ? -1L : Long.valueOf(row.get(10).toString()));
					bean.setFprq(null == row.get(11) ? null : row.getTimestamp(11).getTime());
					bean.setYmtzfwq(null == row.get(12) ? -1 : Integer.valueOf(row.get(12).toString()));
					bean.setDwmc(null == row.get(13) ? "N/A" : row.getString(13));
					bean.setDwxz(null == row.get(14) ? -1L : Long.valueOf(row.get(14).toString()));
					bean.setDwxzjb(null == row.get(15) ? -1L : Long.valueOf(row.get(15).toString()));
					bean.setDwfl(null == row.get(16) ? -1L : Long.valueOf(row.get(16).toString()));
					bean.setDwhyfl(null == row.get(17) ? -1L : Long.valueOf(row.get(17).toString()));
					bean.setJyxkzh(null == row.get(18) ? "N/A" : row.getString(18));
					bean.setShengid(null == row.get(19) ? -1L : Long.valueOf(row.get(19).toString()));
					bean.setShiid(null == row.get(20) ? -1L : Long.valueOf(row.get(20).toString()));
					bean.setXianid(null == row.get(21) ? -1L : Long.valueOf(row.get(21).toString()));
					bean.setXxdz(null == row.getString(22) ? "N/A" : row.getString(22));
					bean.setLxrxm(null == row.getString(23) ? "N/A" : row.getString(23));
					bean.setLxrdh(null == row.getString(24) ? "N/A" : row.getString(24));
					bean.setLxrdzyj(null == row.get(25) ? "N/A" : row.getString(25));
					bean.setWgip(null == row.get(26) ? -1L : Long.valueOf(row.get(26).toString()));
					bean.setWgdz(null == row.get(27) ? "N/A" : row.getString(27));
					bean.setBz(null == row.get(28) ? "N/A" : row.getString(28));
					bean.setIplx(null == row.get(29) ? -1 : Integer.valueOf(row.get(29).toString()));
				}
				return bean;
			}
		});
		return rdd;
	}
	
	
	
	
	
	/**
	 * 获取当前行号
	 * @return
	 */
	public static String getLineInfo()
    {
        StackTraceElement ste = new Throwable().getStackTrace()[1];
        return ste.getFileName() + ": Line " + ste.getLineNumber();
    }
	
	
	
	
	
	/** 第一阶段:保留合法ip 
	 * 以下方法暂不用*/
	// 读取BR2002表内容getElement_br2002
	private JavaRDD<IpBean> BR2002(DataFrame dataSource, IpBean newiB) {
		System.out.println("读取br2002表内容...");
		JavaRDD<IpBean> BR2002_RDD=getElement_br2002(dataSource).map(new Function<Map<String,Object>, IpBean>() {

			private static final long serialVersionUID = 1L;

			@Override
			public IpBean call(Map<String, Object> map) throws Exception {
				IpBean newiB = new IpBean();
				newiB.setId(Long.parseLong(String.valueOf(map.get("id"))));
				newiB.setQsip(null == map.get("qsip") ? null : Long.parseLong(String.valueOf(map.get("qsip"))));
				newiB.setZzip(null == map.get("zzip") ? null : Long.parseLong(String.valueOf(map.get("zzip"))));
				newiB.setBbdw_id(null == map.get("bbdw_id") ? null : Long.parseLong(String.valueOf(map.get("bbdw_id"))));
				/*Long syfs = Long.parseLong(String.valueOf(map.get("syfs")));
				if (syfs != null && !"".equals(syfs)) {
					newiB.setSyfs(syfs);
				}*/
				newiB.setSyfs(null == map.get("syfs") ? null : Long.parseLong(String.valueOf(map.get("syfs"))));
				/*Long shengid = Long.parseLong(String.valueOf(map.get("shengid")));
				if (shengid != null && !"".equals(shengid)) {
					newiB.setShengid(shengid);
				}*/
				newiB.setShengid(null == map.get("shengid") ? null : Long.parseLong(String.valueOf(map.get("shengid"))));
				/*Long shiid = Long.parseLong(String.valueOf(map.get("shiid")));
				if (shiid != null && !"".equals(shiid)) {
					newiB.setShiid(shiid);
				}*/
				newiB.setShiid(null == map.get("shiid") ? null : Long.parseLong(String.valueOf(map.get("shiid"))));
				/*Long xianid = Long.parseLong(String.valueOf(map.get("xianid")));
				if (xianid != null && !"".equals(xianid)) {
					newiB.setXianid(xianid);
				}*/
				newiB.setXianid(null == map.get("xianid") ? null : Long.parseLong(String.valueOf(map.get("xianid"))));
				newiB.setHouseCode(null == map.get("houseCode") ? null : String.valueOf(map.get("houseCode")));
				newiB.setHouseName(null == map.get("houseName") ? null : String.valueOf(map.get("houseName")));
				
				/*Long fpfs = Long.parseLong(String.valueOf(map.get("fpfs")));
				if (fpfs != null && !"".equals(fpfs)) {
					newiB.setFpfs(fpfs);
				}*/
				newiB.setFpfs(null == map.get("fpfs") ? null : Long.parseLong(String.valueOf(map.get("fpfs"))));
				newiB.setDwmc(String.valueOf(map.get("dwmc")));
				newiB.setLxrxm(String.valueOf(map.get("lxrxm")));
				newiB.setLxrdh(String.valueOf(map.get("lxrdh")));
				/*Long dwxz = Long.parseLong(String.valueOf(map.get("dwxz")));
				if (dwxz != null && !"".equals(dwxz)) {
					newiB.setDwxz(dwxz);
				}*/
				newiB.setDwxz(null==map.get("dwxz")? null :Long.parseLong(String.valueOf(map.get("dwxz"))));
				/*Long dwhyfl = Long.parseLong(String.valueOf(map.get("dwhyfl")));
				if (dwhyfl != null && !"".equals(dwhyfl)) {
					newiB.setDwhyfl(dwhyfl);
				}*/
				newiB.setDwhyfl(null==map.get("dwhyfl")? null :Long.parseLong(String.valueOf(map.get("dwhyfl"))));
				newiB.setBz(String.valueOf(map.get("bz")));
				newiB.setSourceIdcId(String.valueOf(map.get("sourceIdcId")));
				newiB.setOperatorsId(String.valueOf(map.get("operatorsId")));
				newiB.setOperatorsName(String.valueOf(map.get("operatorsName")));
				newiB.setOperatorsBh(String.valueOf(map.get("operatorsBh")));
				newiB.setXxdz(String.valueOf(map.get("xxdz")));
				// 将科学计数法转化正int，（Oracle中number类型获取到的列值值科学计数法表示的）
				BigDecimal bd = new BigDecimal(String.valueOf(map.get("jtiplx")));
				int jtiplx = (int) Double.parseDouble(bd.toPlainString());
				newiB.setJtiplx(jtiplx);
				newiB.setStorageType(String.valueOf(Math.round(Double.valueOf(String.valueOf(map.get("storageType"))))));
				return newiB;
			}
		});
		
		return BR2002_RDD;
	}

	
	// 读取IP_GJCX_BAXX表内容
	private JavaRDD<IpBean> IP_GJCX_BAXX(DataFrame dataSource,IpBean newiB) throws Exception {
		System.out.println("读取IP_GJCX_BAXX表内容...");
		JavaRDD<IpBean> IP_GJCX_BAXX_RDD = getElement_baxx(dataSource).map(new Function<Map<String,Object>, IpBean>() {

			private static final long serialVersionUID = 1L;

			@Override
			public IpBean call(Map<String, Object> map) throws Exception {
				IpBean newiB = new IpBean();
				String storageType = String.valueOf((Math.round(Double.valueOf(String.valueOf(map.get("storageType"))))));
				newiB.setStorageType(storageType);
				String tflag = String.valueOf((Math.round(Double.valueOf(String.valueOf(map.get("tflag"))))));
				newiB.setTflag(tflag);
				newiB.setId(Long.parseLong(String.valueOf(map.get("id"))));

				if (tflag.equals(IpCheckConstants.IP_ALLOCATE_TYPE)) {
					newiB.setBbdw_id(Long.parseLong(String.valueOf(map.get("bbdw_id"))));

					Date sbsj_s = (Date) map.get("sbsj");
					Date fprq_s = (Date) map.get("fprq");
					Long sbsj = sbsj_s.getTime();
					newiB.setSbsj(sbsj);
					
					Long fprq_l = 0L;
					if(fprq_s==null){
						fprq_l = null;
					}else{
						fprq_l = (long) fprq_s.getDate();
					}
					
					newiB.setQsip(Long.parseLong(String.valueOf(map.get("qsip"))));
					newiB.setZzip(Long.parseLong(String.valueOf(map.get("zzip"))));
					// 将科学计数法转化正int，（Oracle中number类型获取到的列值值科学计数法表示的）
					BigDecimal bd = new BigDecimal(String.valueOf(map.get("sfgn")));
					Long sfgn = (long) Double.parseDouble(bd.toPlainString());
					newiB.setSfgn(sfgn);
					newiB.setFpfs((Math.round(Double.valueOf(String.valueOf(map.get("fpfs"))))));
					newiB.setFpdx((Math.round(Double.valueOf(String.valueOf(map.get("fpdx"))))));
					newiB.setSyfs((Math.round(Double.valueOf(String.valueOf(map.get("syfs"))))));
					if (fprq_l != null) {
						Long fprq = fprq_s.getTime();// Date类型
						if (fprq == null) {
							fprq = -1L;
						}
						newiB.setFprq(fprq);
					}
					newiB.setYmtzfwq((int) (Math.round(Double.valueOf(String
							.valueOf(map.get("wgip"))))));
					newiB.setDwmc(String.valueOf(map.get("dwmc")));
					newiB.setDwxz((Math.round(Double.valueOf(String.valueOf(map
							.get("dwxz"))))));
					newiB.setDwxzjb((Math.round(Double.valueOf(String
							.valueOf(map.get("dwxzjb"))))));
					newiB.setDwfl((Math.round(Double.valueOf(String.valueOf(map
							.get("dwfl"))))));
					newiB.setDwhyfl((Math.round(Double.valueOf(String
							.valueOf(map.get("dwhyfl"))))));
					newiB.setJyxkzh(String.valueOf(map.get("jyxkzh")));
					newiB.setShengid((Math.round(Double.valueOf(String
							.valueOf(map.get("shengid"))))));
					newiB.setShiid((Math.round(Double.valueOf(String
							.valueOf(map.get("shiid"))))));
					newiB.setXianid((Math.round(Double.valueOf(String
							.valueOf(map.get("xianid"))))));
					newiB.setXxdz(String.valueOf(map.get("xxdz")));
					newiB.setLxrxm(String.valueOf(map.get("lxrxm")));
					newiB.setLxrdh(String.valueOf(map.get("lxrdh")));
					newiB.setLxrdzyj(String.valueOf(map.get("lxrdzyj")));
					newiB.setWgip((Math.round(Double.valueOf(String.valueOf(map
							.get("wgip"))))));
					newiB.setWgdz(String.valueOf(map.get("wgdz")));
					newiB.setBz(String.valueOf(map.get("bz")));
					// 将科学计数法转化正int，（Oracle中number类型获取到的列值值科学计数法表示的）
					BigDecimal bd1 = new BigDecimal(String.valueOf(map.get("iplx")));
					int iplx = (int) Double.parseDouble(bd1.toPlainString());
					newiB.setIplx(iplx);

				}

				return newiB;
			}
		});
	
		JavaRDD<IpBean> resultRDD=iPCheckZjUtil.getFilterIpBeanRdd(IP_GJCX_BAXX_RDD);
		return resultRDD;
	}
	
	//获取备案表sql
	private JavaRDD<Map<String, Object>> getElement_baxx(DataFrame dataSource) {
			JavaRDD<Map<String, Object>> rdd = dataSource.toJavaRDD().map(new Function<Row, Map<String, Object>>() {

						private static final long serialVersionUID = 1L;

						@Override
						public Map<String, Object> call(Row row) throws Exception {
							Map<String, Object> map = new HashMap<String, Object>();
							try{
								// 日期转化
								Date sbsj_date = row.getAs("SBSJ");
								Date fprq_date = row.getAs("FPRQ");
								map.put("storageType", row.getAs("STORAGE_TYPE"));
								map.put("tflag", row.getAs("FT_FLAG"));
								map.put("id", row.getAs("ID"));
								map.put("bbdw_id", row.getAs("BBDW_ID"));
								map.put("sbsj", sbsj_date);// Date类型
								map.put("qsip", row.getAs("QSIP"));
								map.put("zzip", row.getAs("ZZIP"));
								map.put("sfgn", row.getAs("SFGN"));
								map.put("fpfs", row.getAs("FPFS"));
								map.put("fpdx", row.getAs("FPDX"));
								map.put("syfs", row.getAs("SYFS"));
								map.put("fprq", fprq_date);// Date类型
								map.put("ymtzfwq", row.getAs("YMTZFWQ"));
								map.put("dwmc", row.getAs("DWMC"));
								map.put("dwxz", row.getAs("DWXZ"));
								map.put("dwxzjb", row.getAs("DWXZJB"));
								map.put("dwfl", row.getAs("DWFL"));
								map.put("dwhyfl", row.getAs("DWHYFL"));
								map.put("jyxkzh", row.getAs("JYXKZH"));
								map.put("shengid", row.getAs("SHENGID"));
								map.put("shiid", row.getAs("SHIID"));
								map.put("xianid", row.getAs("XIANID"));
								map.put("xxdz", row.getAs("XXDZ"));
								map.put("lxrxm", row.getAs("LXRXM"));
								map.put("lxrdh", row.getAs("LXRDH"));
								map.put("lxrdzyj", row.getAs("LXRDZYJ"));
								map.put("wgip", row.getAs("WGIP"));
								map.put("wgdz", row.getAs("WGDZ"));
								map.put("bz", row.getAs("BZ"));
								map.put("iplx", row.getAs("IP_LX"));
							}catch(Exception e){
								System.err.println("查询备案表数据err:"+iPCheckZjUtil.getLineInfo());
							}
							return map;
						}
					});
			return rdd;
		}

	//获取BR2002表sql
	private JavaRDD<Map<String, Object>> getElement_br2002(DataFrame dataSource) {
			JavaRDD<Map<String, Object>> rdd = dataSource.toJavaRDD().map(new Function<Row, Map<String, Object>>() {

						private static final long serialVersionUID = 1L;

						@Override
						public Map<String, Object> call(Row row) throws Exception {
							
							Map<String,  Object> map = new HashMap<String, Object>();
							try{   
								map.put("id", row.getAs("ID"));
								map.put("qsip", row.getAs("BENGIN_IP_VALUE"));
								map.put("zzip", row.getAs("END_IP_VALUE"));
								map.put("bbdw_id", row.getAs("ISP_ID"));
								map.put("syfs", row.getAs("IP_TYPE"));
								map.put("shengid", row.getAs("PROVINCE"));
								map.put("shiid", row.getAs("CITY"));
								map.put("xianid", row.getAs("COUNTY"));
								map.put("houseCode", row.getAs("HOUSE_CODE"));
								map.put("houseName", row.getAs("HOUSE_NAME"));
								map.put("fpfs", row.getAs("IP_USED_TYPE"));
								map.put("dwmc", row.getAs("USING_UNIT_NAME"));
								map.put("lxrxm", row.getAs("USING_UNIT_LINKMAN"));
								map.put("lxrdh", row.getAs("USING_UNIT_TEL"));
								map.put("dwxz", row.getAs("USING_UNIT_TYPE"));
								map.put("dwhyfl", row.getAs("USING_UNIT_INDUSTRY_TYPE"));
								map.put("bz", row.getAs("REMARK"));
								map.put("sourceIdcId", row.getAs("SOURCE_UNIT"));
								map.put("operatorsId", row.getAs("OPERATORS_ID"));
								map.put("operatorsName", row.getAs("OPERATORS_NAME"));
								map.put("operatorsBh", row.getAs("OPERATORS_JYXKZBH"));
								map.put("xxdz", row.getAs("ADDR"));
								map.put("jtiplx", row.getAs("JTIPLX"));
								map.put("storageType", row.getAs("STORAGE_TYPE"));
							}catch(Exception e){
								System.err.println("查询br2002表数据err:"+iPCheckZjUtil.getLineInfo());
							}
							return map;
						}
					});
			return rdd;
		}


}
